package com.k.activemq.openwire2;

import com.k.activemq.ActiveMQUtils;
import com.k.activemq.MqConnectionEntity;

import javax.jms.*;

public class MqListener {
    public static void main(String[] args) {
//        MqConnectionEntity connectionEntity = ActiveMQUtils.getConnectionData();
        MqConnectionEntity connectionEntity = MqConnectionEntity.newBuilder()
                .setUrl("tcp://192.168.241.89:22000")  // brokerURL
                .setUsername("admin")             // username
                .setPassword("Command_123")             // password
                .setDestination("end_report")      // Topic or Queue Name
                .setPubSubDomain(true)            // true: Topic
                .setMessage("hello")              // Message
                .setByteMessage(true)             // true: ByteMessage
                .build();
        System.out.println("mq listener:"+connectionEntity.toString());
        try {
            System.out.println(Thread.currentThread().getId()+"Mq start listener1 ..."+connectionEntity.getPubSubDomain()+":"+connectionEntity.getDestination());
            listener(connectionEntity);
            System.out.println(Thread.currentThread().getId()+"Mq end listener1");
        } catch (JMSException e) {
            e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getId()+"Mq listener1 finished!");
    }
    public static void listener(MqConnectionEntity connEntity) throws JMSException {
        // 创建连接工厂
        ConnectionFactory connectionFactory = ActiveMQUtils.getActiveMQConnectionFactory(connEntity.getUrl());

        // 创建连接
        Connection connection = connectionFactory.createConnection(connEntity.getUsername(), connEntity.getPassword());
//        Connection connection = connectionFactory.createConnection();
        // 打开连接
        connection.start();
        // 创建会话
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 创建队列目标,并标识队列名称，消费者根据队列名称接收数据
        Destination destination = connEntity.getPubSubDomain() ? session.createTopic(connEntity.getDestination()) : session.createQueue(connEntity.getDestination());
        session.createTopic(connEntity.getDestination());
        // 创建消费者
        MessageConsumer consumer = session.createConsumer(destination);
        // 创建消费的监听 主线程非阻塞
        receive(consumer);
        System.out.println(Thread.currentThread()+"listener2() finished!");
    }
    private static void receiveAsync(MessageConsumer consumer) throws JMSException {
        consumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                System.out.println(Thread.currentThread()+": MyListener2:"+message);
                if( message instanceof  TextMessage ) {
                    TextMessage textMessage = (TextMessage) message;
                    try {
                        String msgStr = textMessage.getText();
                        System.out.println(ActiveMQUtils.formatMsg("TextMessage",Thread.currentThread().toString(),msgStr));
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }else if(message instanceof BytesMessage){
                    BytesMessage byteMessage = (BytesMessage) message;
                    try {
                        String msgStr = ActiveMQUtils.byteMessageToString(byteMessage);
                        System.out.println(ActiveMQUtils.formatMsg("BytesMessage",Thread.currentThread().toString(),msgStr));
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }else if(message instanceof MapMessage){
                    MapMessage mapMessage = (MapMessage) message;
                    String msgStr = mapMessage.toString();
                    System.out.println(ActiveMQUtils.formatMsg("MapMessage",Thread.currentThread().toString(),msgStr));
                }else if(message instanceof ObjectMessage){
                    String msgStr = "";
                    System.out.println(ActiveMQUtils.formatMsg("ObjectMessage",Thread.currentThread().toString(),msgStr));
                }else if(message instanceof StreamMessage){
                    String msgStr = "";
                    System.out.println(ActiveMQUtils.formatMsg("StreamMessage",Thread.currentThread().toString(),msgStr));
                }else{
                    System.out.println("消息类型：Unknown");
                }
            }
        });
    }
    private static void receive(MessageConsumer consumer) throws JMSException {
        while(true) {
            Message msg = consumer.receive();
            if( msg instanceof  TextMessage ) {
                TextMessage textMessage = (TextMessage) msg;
                try {
                    String msgStr = textMessage.getText();
                    System.out.println(ActiveMQUtils.formatMsg("TextMessage",Thread.currentThread().toString(),msgStr));
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            } else if(msg instanceof BytesMessage){
                BytesMessage byteMessage = (BytesMessage) msg;
                try {
                    String msgStr = ActiveMQUtils.byteMessageToString(byteMessage);
                    System.out.println(ActiveMQUtils.formatMsg("BytesMessage",Thread.currentThread().toString(),msgStr));
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            } else {
                System.out.println("Received msg:NoHandler");
            }
        }
    }
}